#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>

#define MAX_TBF 1024
#define min(x,y) ((x < y) ? x : y)

struct mytbf_st
{
    int cps;
    int burst;
    int token;
    int pos;
    pthread_mutex_t mtx;
    pthread_cond_t cond;
};

struct mytbf_st *jobs[MAX_TBF];

static volatile int initialized = 0;
static  pthread_t scheduler;

static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;

static void *background(void *any)
{
    struct timespec tm;
    tm.tv_sec = 1;
    while (1)
    {
        sleep(1);
        pthread_mutex_lock(&mtx);

        for (int i = 0; i < MAX_TBF; i++)
        {
            struct mytbf_st *p = jobs[i];
            if (p == NULL)
            {
                continue;
            }
            pthread_mutex_lock(&p->mtx);
            p->token += p->cps;
            p->token = min(p->token, p->burst);
            pthread_mutex_unlock(&p->mtx);
            pthread_cond_broadcast(&p->cond);
        }
        pthread_mutex_unlock(&mtx);
    }
}

int mytbf_destroy_unlocked(int id)
{
    struct mytbf_st *p = jobs[id];
    jobs[id] = NULL;
    if (p != NULL)
    {
        pthread_cond_destroy(&p->cond);
        pthread_mutex_destroy(&p->mtx);
        free(p);
        return id;
    }
    return -1;
}

int mytbf_destroy(int id)
{
    pthread_mutex_lock(&mtx);
    id = mytbf_destroy_unlocked(id);
    pthread_mutex_unlock(&mtx);
    return id;
}

static int get_pos_unlocked()
{
    int i;
    for (i = 0; i < MAX_TBF; i++)
    {
        if (jobs[i] == NULL)
            return i;
    }
    return -1;
}

static int mytbf_fetchtoken(int id, int token)
{
    struct mytbf_st* p = jobs[id];
    if (p == NULL)
    {
        return -1;
    }

    pthread_mutex_lock(&p->mtx);
    while(p->token <= 0) {
        // 在这里等待通知
        // wait 会被 broadcast 打断，但不会解锁
        pthread_cond_wait(&p->cond, &p->mtx);
    }

    int n = min(token, p->token);
    p->token -= n;
    pthread_mutex_unlock(&p->mtx);
    return n;
}

static int mytbf_returntoken(int id, int token)
{
    struct mytbf_st* p = jobs[id];
    if (p == NULL)
    {
        return -1;
    }

    pthread_mutex_lock(&p->mtx);
    p->token += token;
    p->token = min(p->token, p->burst);
    pthread_mutex_unlock(&p->mtx);
    pthread_cond_broadcast(&p->cond);
}


static int mytbf_init(int cps, int burst)
{
    pthread_mutex_lock(&mtx);

    if (!initialized)
    {
        pthread_create(&scheduler, NULL, background, NULL);
        initialized = 1;
    }

    int i = get_pos_unlocked();
    if (i < 0)
    {
        pthread_mutex_unlock(&mtx);
        return i;
    }

    struct mytbf_st *p = malloc(sizeof(struct mytbf_st));
    p->cps = cps;
    p->burst = burst;
    pthread_cond_init(&p->cond, NULL);
    pthread_mutex_init(&p->mtx, NULL);
    jobs[i] = p;
    pthread_mutex_unlock(&mtx);
    return i;
}

static int mytbf_unload()
{
    pthread_mutex_lock(&mtx);
    for(int i = 0; i < MAX_TBF; i++)
        mytbf_destroy_unlocked(i);

    pthread_mutex_unlock(&mtx);
    pthread_mutex_destroy(&mtx);
    pthread_cancel(scheduler);
    pthread_join(scheduler, NULL);
}

int main(int argc, char** argv)
{   
    if (argc < 2) {
        fprintf(stderr, "%s filename\n", argv[0]);
        exit(1);
    }

    int id = mytbf_init(10, 100);
    char buf[256];

    int fd = open(argv[1], O_RDONLY);

    while(1) {
        int fetched =  mytbf_fetchtoken(id, 256);
        ssize_t sz = read(fd, buf, fetched);

        if(sz == 0)
            break;

        write(1, buf, sz);
        mytbf_returntoken(id,  fetched - sz);

    }

    mytbf_unload();
    close(fd);
}